package com.zx.notification.consumer.listener;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.zx.notification.common.enums.MqTypeEnum;
import com.zx.notification.common.event.ConfigReloadEvent;
import com.zx.notification.common.model.NotificationChannelsConfig;
import com.zx.notification.consumer.config.ActiveMqConsumerConfig;
import com.zx.notification.consumer.config.ChannelConfigManager;
import com.zx.notification.consumer.service.IMessageProcessService;
import com.zx.notification.consumer.util.MessageProcessUtil;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import jakarta.jms.ConnectionFactory;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.event.EventListener;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
import org.springframework.jms.listener.MessageListenerContainer;
import org.springframework.jms.listener.adapter.MessageListenerAdapter;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;

/**
 * ActiveMQ消息监听器
 * 用于处理ActiveMQ通知消息
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class ActiveMqMsgListener {

    private final IMessageProcessService messageProcessService;
    private final ChannelConfigManager channelConfigManager;
    private final ActiveMqConsumerConfig activeMqConfig;
    private final ConnectionFactory connectionFactory;
    // 存储动态创建的监听器容器
    private final Map<String, MessageListenerContainer> listenerContainers = new ConcurrentHashMap<>();
    // 存储渠道类型与目标名称的映射关系
    private final Map<String, String> channelDestinationMap = new ConcurrentHashMap<>();
    // 重入锁防止并发重载冲突
    private final ReentrantLock reloadLock = new ReentrantLock();

    private final ObjectMapper objectMapper = new ObjectMapper();

    /**
     * 初始化监听器，从数据库加载启用的ActiveMQ渠道配置
     */
    @PostConstruct
    public void init() {
        if (!activeMqConfig.isEnabled()) {
            log.info("ActiveMQ未启用，跳过ActiveMQ监听器初始化");
            return;
        }
        loadActiveActiveMqChannelConfigurations();
    }

    /**
     * 从数据库加载启用的ActiveMQ渠道配置
     */
    private void loadActiveActiveMqChannelConfigurations() {
        try {
            // 查询所有启用的ActiveMQ渠道配置
            List<NotificationChannelsConfig> channelConfigs = channelConfigManager.getActiveChannelConfigsByMqType(MqTypeEnum.ACTIVEMQ.getCode());

            int successCount = 0;
            for (NotificationChannelsConfig config : channelConfigs) {
                try {
                    // 添加队列名称 - 添加空值检查
                    String queueName = config.getQueueName();
                    if (queueName != null && !queueName.trim().isEmpty()) {
                        channelDestinationMap.put(config.getChannelType() + "_QUEUE", queueName.trim());
                        registerListenerForDestination(queueName.trim());
                        log.info("加载启用的ActiveMQ渠道配置: {} -> 队列:{}", config.getChannelType(), queueName.trim());
                        successCount++;
                    }

                    // 添加主题名称 - 添加空值检查
                    String topicName = config.getTopicName();
                    if (topicName != null && !topicName.trim().isEmpty()) {
                        channelDestinationMap.put(config.getChannelType() + "_TOPIC", topicName.trim());
                        registerListenerForDestination(topicName.trim());
                        log.info("加载启用的ActiveMQ渠道配置: {} -> 主题:{}", config.getChannelType(), topicName.trim());
                        successCount++;
                    }
                } catch (Exception e) {
                    log.error("加载渠道配置失败: {}", config.getChannelType(), e);
                }
            }
            log.info("ActiveMQ渠道配置加载完成，成功加载 {} 个监听器", successCount);
        } catch (Exception e) {
            log.error("加载启用的ActiveMQ渠道配置失败", e);
        }
    }

    /**
     * 为指定目标注册监听器
     *
     * @param destination 目标名称（队列或主题）
     */
    private void registerListenerForDestination(String destination) {
        if (listenerContainers.containsKey(destination)) {
            log.debug("监听器已存在: {}", destination);
            return;
        }

        DefaultMessageListenerContainer container = null;
        try {
            container = createMessageListenerContainer(destination);
            // 显式初始化
            container.afterPropertiesSet();
            // 启动监听器容器
            container.start();

            // 只有启动成功后才放入Map
            listenerContainers.put(destination, container);

            log.info("成功为目标 {} 注册监听器", destination);
        } catch (Exception e) {
            log.error("为目标 {} 注册监听器失败", destination, e);
            // 清理资源：如果容器创建成功但启动失败，需要销毁
            if (container != null) {
                try {
                    if (container.isRunning()) {
                        container.stop();
                    }
                    container.destroy();
                } catch (Exception ex) {
                    log.warn("清理失败容器资源时发生错误: {}", destination, ex);
                }
            }
        }
    }

    /**
     * 创建消息监听器容器
     *
     * @param destination 目标名称
     * @return 消息监听器容器
     */
    private DefaultMessageListenerContainer createMessageListenerContainer(String destination) {
        // 创建消息监听器容器
        DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setDestinationName(destination);

        // 创建消息监听器适配器
        MessageListenerAdapter listenerAdapter = new MessageListenerAdapter();
        listenerAdapter.setDelegate(this);
        listenerAdapter.setDefaultListenerMethod("handleSingleMessage");
        container.setMessageListener(listenerAdapter);
        return container;
    }

    /**
     * 处理ActiveMQ通知消息
     *
     * @param message 通知消息（JSON格式字符串）
     */
    public void handleSingleMessage(String message) {
        if (!activeMqConfig.isEnabled()) {
            log.debug("ActiveMQ未启用，忽略接收到的消息");
            return;
        }

        if (message == null || message.trim().isEmpty()) {
            log.warn("接收到空消息，忽略处理");
            return;
        }

        String substring = message.substring(0, Math.min(message.length(), 200));
        try {
            MessageProcessUtil.processMessage(message, MqTypeEnum.ACTIVEMQ, objectMapper, channelConfigManager, messageProcessService);
            log.debug("成功处理ActiveMQ消息: {}", substring);
        } catch (Exception e) {
            log.error("处理ActiveMQ消息失败: {}", substring, e);
            // 根据业务需求决定是否抛出异常
            // 如果抛出异常，ActiveMQ会进行重试
            throw new RuntimeException("消息处理失败", e);
        }
    }

    /**
     * 重新加载渠道配置
     */
    public void reloadChannelConfigurations() {
        if (!activeMqConfig.isEnabled()) {
            log.debug("ActiveMQ未启用，跳过重新加载渠道配置");
            return;
        }

        // 使用锁确保同一时间只有一个重载操作
        if (!reloadLock.tryLock()) {
            log.warn("配置重载操作正在进行中，跳过本次请求");
            return;
        }

        try {
            // 创建旧容器副本
            Map<String, MessageListenerContainer> oldContainers = new ConcurrentHashMap<>(listenerContainers);

            // 清空当前状态
            listenerContainers.clear();
            channelDestinationMap.clear();

            // 加载新配置
            loadActiveActiveMqChannelConfigurations();

            // 异步关闭旧容器，避免阻塞重载过程
            CompletableFuture.runAsync(() -> {
                oldContainers.forEach((destination, container) -> {
                    try {
                        if (container instanceof DefaultMessageListenerContainer defaultContainer) {
                            if (defaultContainer.isRunning()) {
                                defaultContainer.stop();
                            }
                            defaultContainer.destroy();
                        }
                        log.debug("成功停止旧监听器: {}", destination);
                    } catch (Exception e) {
                        log.warn("停止旧监听器时发生错误: {}", destination, e);
                    }
                });
                log.info("异步关闭 {} 个旧监听器完成", oldContainers.size());
            });

        } catch (Exception e) {
            log.error("配置重载失败", e);
        } finally {
            reloadLock.unlock();
        }
    }

    /**
     * 监听配置重载事件
     *
     * @param event 配置重载事件
     */
    @EventListener
    public void handleConfigReloadEvent(ConfigReloadEvent event) {
        if (!activeMqConfig.isEnabled()) {
            log.debug("ActiveMQ未启用，忽略配置重载事件");
            return;
        }

        log.info("收到配置重载事件: {}", event.getMessage());
        // 重新加载渠道配置
        reloadChannelConfigurations();
    }

    /**
     * 优雅关闭，防止资源泄漏
     */
    @PreDestroy
    public void destroy() {
        log.info("开始关闭ActiveMQ消息监听器...");
        listenerContainers.forEach((destination, container) -> {
            try {
                if (container instanceof DefaultMessageListenerContainer defaultContainer) {
                    if (defaultContainer.isRunning()) {
                        defaultContainer.stop();
                    }
                    defaultContainer.destroy();
                }
                log.debug("成功停止监听器: {}", destination);
            } catch (Exception e) {
                log.warn("停止监听器时发生错误: {}", destination, e);
            }
        });
        listenerContainers.clear();
        channelDestinationMap.clear();
        log.info("ActiveMQ消息监听器关闭完成");
    }
}